package rabbit.flt.test.webflux;

import junit.framework.TestCase;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rabbit.flt.common.Headers;
import rabbit.flt.common.Traceable;
import rabbit.flt.common.context.TraceContext;
import rabbit.flt.common.exception.FltException;
import rabbit.flt.common.trace.TraceData;
import rabbit.flt.test.common.spi.TestTraceHandler;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.ByteBufFlux;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpClientResponse;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.function.BiFunction;

public class ReactorHttpCase {

    private Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 单一请求测试
     *
     * @throws Exception
     */
    public void singleTraceableCase() throws Exception {
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });

        singleTraceableCall();

        semaphore.acquire(5);
        TestCase.assertNotNull(map.get("0"));
        TestCase.assertEquals("singleTraceableCall", map.get("0").getNodeName());
        TestCase.assertNotNull(map.get("0-0"));
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertEquals("simpleTrace", map.get("0-1").getNodeName());
        TestCase.assertTrue(map.get("0").getTraceId().equals(map.get("0-0").getTraceId()));
        TestCase.assertTrue(map.get("0-0-0-0").getNodeName().equals("hello"));
        TestCase.assertTrue(map.get("0-0-0").getNodeName().equals("/child/hello"));
        TestCase.assertFalse(TraceContext.isTraceOpened());
        TestTraceHandler.setDiscardDataHandler(null);
    }


    /**
     * 使用responseConnection方法处理
     *
     * @throws Exception
     */
    public void responseConnectionCase() throws Exception {
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        responseConnectionCall();
        semaphore.acquire(5);
        TestCase.assertNotNull(map.get("0"));
        TestCase.assertEquals("responseConnectionCall", map.get("0").getNodeName());
        TestCase.assertNotNull(map.get("0-0"));
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertEquals("simpleTrace", map.get("0-1").getNodeName());
        TestCase.assertTrue(map.get("0").getTraceId().equals(map.get("0-0").getTraceId()));
        TestCase.assertTrue(map.get("0-0-0-0").getNodeName().equals("hello"));
        TestCase.assertTrue(map.get("0-0-0").getNodeName().equals("/mvc/hello"));
        TestCase.assertFalse(TraceContext.isTraceOpened());
        TestTraceHandler.setDiscardDataHandler(null);
    }

    private void logTraceData(String s, TraceData d) {
        logger.info(s, d.getNodeName(), d.getSpanId());
    }

    @Traceable
    private void responseConnectionCall() throws Exception {
        Semaphore semaphore = new Semaphore(0);
        HttpClient httpClient = HttpClient.create();
        HttpClient.ResponseReceiver<?> receiver = httpClient.get().uri("http://localhost:8888/mvc/hello");
        receiver.responseConnection((httpClientResponse, connection) -> {
            Mono<List<String>> collectList = connection.inbound().receive().asByteArray()
                    .map(String::new).collectList();
            return collectList.map(list -> {
                logger.info("responseConnectionCall: {}", list.get(0));
                semaphore.release();
                return list;
            });
        }).collectList().block();
        semaphore.acquire();
        simpleTrace();
    }

    @Traceable
    private void singleTraceableCall() throws Exception {
        Semaphore semaphore = new Semaphore(0);
        HttpClient httpClient = HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999"));
        HttpClient.ResponseReceiver<?> receiver = httpClient.get().uri("http://localhost:8888/child/hello");
        BiFunction<HttpClientResponse, ByteBufFlux, Publisher<String>> biFunction = (httpClientResponse, byteBufFlux) -> {
            Flux<byte[]> flux = byteBufFlux.asByteArray();
            return flux.map(String::new);
        };
        receiver.response(biFunction).collectList().subscribe(s -> {
            logger.info("response: {}", s.get(0));
            semaphore.release();
        });
        semaphore.acquire();
        simpleTrace();
    }

    @Traceable
    private void simpleTrace() {

    }


    /**
     * 调用抛异常的接口，验证error body
     *
     * @throws Exception
     */
    public void responseConnectionErrorCallTest() throws Exception {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        String result = responseConnectionErrorCall();
        semaphore.acquire(4);
        TestCase.assertTrue(result.equals("error"));
        TestCase.assertEquals("responseConnectionErrorCall", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertEquals("error", map.get("0-0").getHttpResponse().getBody());
        TestCase.assertEquals("/mvc/error", map.get("0-0-0").getNodeName());
        TestCase.assertEquals("error", map.get("0-0-0").getHttpResponse().getBody());
        TestCase.assertEquals("error", map.get("0-0-0-0").getNodeName());
    }

    @Traceable
    private String responseConnectionErrorCall() {
        try {
            HttpClient.ResponseReceiver<?> receiver = HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999"))
                    .get().uri("http://localhost:8888/mvc/error");
            return receiver.responseConnection((httpClientResponse, connection) -> {
                ByteBufFlux receive = connection.inbound().receive();
                return receive.asByteArray().map(bytes -> {
                    String value = new String(bytes);
                    logger.info("responseConnectionErrorCall: {}", value);
                    return value;
                }).doFinally(s -> connection.dispose());
            }).collectList().block().get(0);
        } catch (Exception e) {
            return e.getMessage();
        }
    }


    /**
     * 级联测试
     *
     * @throws Exception
     */
    public void cascadedTest() throws Exception {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        callHttpContinuously();
        semaphore.acquire(7);
        TestCase.assertEquals("callHttpContinuously", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertEquals("/mvc/hello", map.get("0-0-0").getNodeName());
        TestCase.assertEquals("hello", map.get("0-0-0-0").getNodeName());
        TestCase.assertEquals("/mvc/hello1", map.get("0-1-0").getNodeName());
        TestCase.assertEquals("hello1", map.get("0-1-0-0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-1").getNodeName());
        TestCase.assertFalse(TraceContext.isTraceOpened());
        TestTraceHandler.setDiscardDataHandler(null);
    }

    @Traceable
    private void callHttpContinuously() {
        Mono<String> fm = HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get().uri("http://localhost:8888/mvc/hello")
                .responseSingle((httpClientResponse, byteBufMono) -> byteBufMono.asByteArray().map(String::new))
                .flatMap(f -> HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get().uri("http://localhost:8888/mvc/hello1")
                        .responseSingle((httpClientResponse, byteBufMono) -> byteBufMono.asByteArray().map(String::new)));
        TestCase.assertEquals("abc", fm.block());
    }

    /**
     * 未捕获的异常
     *
     * @throws InterruptedException
     */
    public void unHandledError() throws InterruptedException {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        String result = unHandledErrorCase();
        semaphore.acquire(4);
        TestCase.assertTrue(result.contains("500"));
        TestCase.assertEquals("unHandledErrorCase", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertTrue(map.get("0-0").getHttpResponse().getBody().contains("500"));
        TestCase.assertEquals("/mvc/unHandledError", map.get("0-0-0").getNodeName());
        TestCase.assertTrue(map.get("0-0").getHttpResponse().getBody().contains("500"));
        TestCase.assertEquals("unHandledError", map.get("0-0-0-0").getNodeName());

        TestTraceHandler.setDiscardDataHandler(null);
    }

    @Traceable
    private String unHandledErrorCase() {
        return HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get().uri("http://localhost:8888/mvc/unHandledError")
                .responseSingle((httpClientResponse, byteBufMono) -> {
                    logger.info("{}", httpClientResponse.status());
                    return byteBufMono.asByteArray().map(bytes -> {
                        logger.info("{}", bytes.length);
                        return new String(bytes);
                    });
                }).block();
    }


    /**
     * 调用抛异常的接口，验证error body
     *
     * @throws Exception
     */
    public void errorCallTest() throws Exception {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        String result = errorCall();
        semaphore.acquire(4);
        TestCase.assertTrue(result.equals("error"));
        TestCase.assertEquals("errorCall", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertEquals("error", map.get("0-0").getHttpResponse().getBody());
        TestCase.assertEquals("/mvc/error", map.get("0-0-0").getNodeName());
        TestCase.assertEquals("error", map.get("0-0-0").getHttpResponse().getBody());
        TestCase.assertEquals("error", map.get("0-0-0-0").getNodeName());

    }

    @Traceable
    private String errorCall() {
        try {
            return HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get().uri("http://localhost:8888/mvc/error")
                    .responseSingle((httpClientResponse, byteBufMono) -> byteBufMono.asByteArray().map(String::new)).block();
        } catch (Exception e) {
            return e.getMessage();
        }
    }


    /**
     * 未处理的异常
     *
     * @throws InterruptedException
     */
    public void offlineTest() throws InterruptedException {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("offlineTest traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        String result = offlineCase();
        TestCase.assertTrue(result.contains("no further information"));
        semaphore.acquire(2);
        TestCase.assertEquals("offlineCase", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        TestCase.assertTrue(map.get("0-0").getHttpResponse().getBody().contains("no further information"));
        TestTraceHandler.setDiscardDataHandler(null);
    }

    public void sseByResponseTest() throws InterruptedException {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("sseByResponseTest traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        sseByResponseCase();
        semaphore.acquire(4);
        TestCase.assertEquals("sseByResponseCase", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        logger.info("sseByResponseCase cost: {}", map.get("0").getCost());
        logger.info("ReactorHttp cost: {}", map.get("0-0").getCost());
        TestCase.assertTrue(map.get("0-0").getCost() >= 3 * 1000);
        TestCase.assertTrue(map.get("0").getCost() >= map.get("0-0").getCost());
        TestCase.assertEquals("/reactor/sse", map.get("0-0-0").getNodeName());
        TestCase.assertEquals("sse", map.get("0-0-0-0").getNodeName());
        TestTraceHandler.setDiscardDataHandler(null);
    }


    public void sseByResponseConnectionTest() throws InterruptedException {
        TestCase.assertFalse(TraceContext.isTraceOpened());
        Map<String, TraceData> map = new ConcurrentHashMap<>();
        Semaphore semaphore = new Semaphore(0);
        TestTraceHandler.setDiscardDataHandler(d -> {
            logTraceData("sseByResponseTest traceData: {}#{}", d);
            map.put(d.getSpanId(), d);
            semaphore.release();
        });
        sseByResponseConnectionCase();
        semaphore.acquire(4);
        TestCase.assertEquals("sseByResponseConnectionCase", map.get("0").getNodeName());
        TestCase.assertEquals("ReactorHttp", map.get("0-0").getNodeName());
        logger.info("sseByResponseConnectionCase cost: {}", map.get("0").getCost());
        TestCase.assertTrue(map.get("0-0").getCost() >= 3 * 1000);
        TestCase.assertTrue(map.get("0").getCost() >= map.get("0-0").getCost());
        TestCase.assertEquals("/reactor/sse", map.get("0-0-0").getNodeName());
        TestCase.assertEquals("sse", map.get("0-0-0-0").getNodeName());
        TestTraceHandler.setDiscardDataHandler(null);
    }


    /**
     * 请求不在的服务
     *
     * @return
     */
    @Traceable
    private String offlineCase() {
        try {
            return HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get().uri("http://localhost:12898/mvc/unHandledError")
                    .responseSingle((httpClientResponse, byteBufMono) -> byteBufMono.asByteArray().map(String::new)).block();
        } catch (Exception e) {
            return e.getMessage();
        }
    }

    @Traceable
    private void sseByResponseCase() {
        try {
            HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get()
                    .uri("http://localhost:8888/reactor/sse")
                    .response((conn, buf) -> {
                        Flux<String> map = buf.asByteArray().map(bytes -> {
                            String data = new String(bytes);
                            logger.info("sse data: {}", data);
                            return data;
                        });
                        return map;
                    }).collectList().block();
        } catch (Exception e) {
            throw new FltException(e);
        }
    }


    @Traceable
    private void sseByResponseConnectionCase() {
        try {
            HttpClient.create().headers(h -> h.set(Headers.SPAN_ID, "9999")).get()
                    .uri("http://localhost:8888/reactor/sse")
                    .responseConnection((resp, conn) -> {
                        Flux<String> map = conn.inbound().receive().asByteArray().map(bytes -> {
                            String s = new String(bytes);
                            logger.info("response connection sse data -> {}", s);
                            return s;
                        });
                        return map;
                    }).collectList().block();
        } catch (Exception e) {
            throw new FltException(e);
        }
    }

}
